package com.rj.data.task.impl;

import com.google.common.collect.Lists;
import com.google.common.io.Files;
import com.google.common.io.LineProcessor;
import com.rj.data.sink.Sink;
import com.rj.data.source.FileSource;
import com.rj.data.source.Source;
import com.rj.data.task.SynTask;
import com.rj.data.task.Task;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;

import java.io.File;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.Arrays;
import java.util.List;

public class CnameFetchTask extends SynTask<List<String>> {

    public CnameFetchTask(Source source, List list) {
        super(source, list);
    }

    @Override
    public Task runTask() throws Exception {
        //TODO 定义具体执行任务
        //读取文件行数，截取企业名称
        Files.asCharSource((FileSource) getSource(), Charset.forName("UTF-8")).readLines(new LineProcessor<List<String>>() {
            List<Sink> sinks = getSinks();
            List<String> lines = Lists.newArrayList();
            List<String> errLines = Lists.newArrayList();
            String[] larr = new String[]{};

            @Override
            public boolean processLine(String line) {
                try {
                    larr = line.split("\\|\\+\\|");
                    if (larr.length < 2) {
                        errLines.add(line);
                        return true;
                    }
                    String cname = larr[2];
                    if (StringUtils.isNotBlank(cname)) {
                        lines.add(cname);
                    }
                    if (lines.size() > 50000) {
                        sinks.forEach(s -> {
                            try {
                                s.writeLines(lines, true, true, null);
                            } catch (IOException e) {

                                throw new RuntimeException("写入文件错误", e);
                            } finally {
                                lines.clear();
                            }
                        });
                    }
                } catch (Exception e) {
                    System.out.println("line====> " + line);

                    System.out.println("larr====> ");
                    Arrays.stream(larr).forEach(x -> System.out.print(x + "<=>"));
                }


                return true;
            }

            @Override
            public List<String> getResult() {
                if (lines.size() > 0) {
                    sinks.forEach(s -> {
                        try {
                            s.writeLines(lines, true, true, null);
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    });
                }
                if (errLines.size() > 0) {

                    try {
                        FileUtils.writeLines(new File("/data/prod/workspace/datacenter/mysql-files/company-base-error.txt"), "UTF-8", errLines, true);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }

                }
                return null;
            }
        });
        return null;
    }
}
